package org.dreamwork.cai.managed;

import org.dreamwork.cai.SRFException;
import org.dreamwork.cai.annotation.ARemoteService;
import org.dreamwork.cai.annotation.ARemoteServiceDiscoverer;
import org.dreamwork.cai.core.IHeartbeatManager;
import org.dreamwork.cai.core.RemoteServiceScanner;
import org.dreamwork.cai.core.SRFBroadcastReceiver;
import org.dreamwork.cai.core.impl.HeartbeatManager;
import org.dreamwork.cai.io.AbstractRouteBroadcaster;
import org.dreamwork.cai.io.AbstractServiceProvider;
import org.dreamwork.cai.core.RemoteServiceFactory;
import org.dreamwork.cai.jmx.CrossApplicationInvocation;
import org.dreamwork.cai.proxy.IServiceResolver;
import org.dreamwork.cai.srf.ServiceRouteFrame;
import org.dreamwork.cai.srf.ServiceRouteTable;
import org.dreamwork.cai.tools.Const;
import org.dreamwork.cai.tools.Misc;
import org.dreamwork.concurrent.broadcast.ILocalBroadcastService;
import org.dreamwork.concurrent.broadcast.LocalBroadcaster;
import org.dreamwork.injection.AConfigured;
import org.dreamwork.injection.IInjectResolvedProcessor;
import org.dreamwork.injection.IObjectContext;
import org.dreamwork.injection.impl.SimpleObjectContext;
import org.dreamwork.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.Arrays;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 远程服务发现程序.
 * <p>负责发现和构建本地 SRF</p>
 */
public abstract class RemoteServiceDiscoverer implements IInjectResolvedProcessor {
    // 受托管对象
    private ServiceRouteTable table;
    private ObjectContextResolver resolver;
    private ObjectContextRemoteServices remoteServices;
    private LocalBroadcaster broadcaster;
    private SRFBroadcastReceiver receiver;
    private CrossApplicationInvocation invocation;
    private HeartbeatManager heartbeat;

    // 本地变量
    private AbstractRouteBroadcaster routeBroadcaster;
    private AbstractServiceProvider provider;

    private final ExecutorService executor = Executors.newCachedThreadPool ();
    private final Logger logger = LoggerFactory.getLogger (RemoteServiceDiscoverer.class);

    @AConfigured ("${cai.router.broadcast.port}")
    private int routeBroadcastPort = 12315;

    @AConfigured ("${cai.router.provider.port}")
    private int providerPort = 12316;

    @AConfigured ("${cai.router.namespace}")
    private String namespace;

    @AConfigured ("${cai.router.heartbeat.scan.interval}")
    private int scanInterval = 1000;

    @AConfigured ("${cai.router.heartbeat.send.interval}")
    private int sendInterval = 10000;

    @AConfigured ("${cai.router.heartbeat.timeout}")
    private int timeout = 60;

    @Resource
    public ServiceRouteTable getServiceRouteTable () {
        return table = new ServiceRouteTable ();
    }

    @Resource
    public IServiceResolver getServiceResolver () {
        return resolver = new ObjectContextResolver ();
    }

    @Resource
    public ObjectContextRemoteServices getObjectContextRemoteServices () {
        return remoteServices = new ObjectContextRemoteServices ();
    }

    @Resource
    public ILocalBroadcastService getLocalBroadcasterService () {
        return broadcaster = new LocalBroadcaster (executor);
    }

    @Resource
    public IHeartbeatManager getHeartbeatManager () {
        return heartbeat = new HeartbeatManager ();
    }

    /**
     * 当远程调用框架准备销魂前暴露的接口，任何清理工作都可以放在这里
     * @throws Throwable 任何错误
     */
    protected void beforeDestroy () throws Throwable {}

    @PreDestroy
    public final void destroy () {
        try {
            beforeDestroy ();
        } catch (Throwable ex) {
            logger.warn (ex.getMessage (), ex);
        }

        // 停止路由广播
        if (routeBroadcaster != null) try {
            routeBroadcaster.unbind ();
        } catch (IOException ex) {
            if (logger.isTraceEnabled ()) {
                logger.warn (ex.getMessage (), ex);
            }
        }

        // 停止路由提供程序
        if (provider != null) try {
            provider.unbind ();
        } catch (IOException ex) {
            if (logger.isTraceEnabled ()) {
                logger.warn (ex.getMessage (), ex);
            }
        }

        // 停止心跳扫描程序
        if (heartbeat != null) {
            heartbeat.shutdown ();
        }

        // 删除本地广播监听器，并停止本地广播
        if (receiver != null && broadcaster != null) {
            broadcaster.unregister (Const.Category.CAT_SRF_EVENT, receiver);
            broadcaster.shutdown ();
        }

        // 注销 CAI JMX 服务
        if (invocation != null) {
            invocation.dispose ();
        }

        // 停止线程池
        executor.shutdownNow ();
    }

    @Override
    public void perform (IObjectContext ctx) throws Exception {
        SimpleObjectContext context = (SimpleObjectContext) ctx;
        ARemoteServiceDiscoverer rsd = getClass ().getAnnotation (ARemoteServiceDiscoverer.class);
        if (rsd != null && !StringUtil.isEmpty (rsd.namespace ())) {
            namespace = rsd.namespace ().trim ();
        }

        int senderPort = (int) (Math.random () * 5534 + 60000);

        if (logger.isTraceEnabled ()) {
            logger.trace ("================ Remote Service Discoverer Bridge =================");
            logger.trace ("= scan.interval     = {}s.", scanInterval / 1000);
            logger.trace ("= send.interval     = {}s.", sendInterval / 1000);
            logger.trace ("= heartbeat.timeout = {}s.", timeout);
            logger.trace ("= broadcast.port    = {}", routeBroadcastPort);
            logger.trace ("= provider.port     = {}", providerPort);
            logger.trace ("= sender.port       = {}", senderPort);
            logger.trace ("= local.namespace   = {}", namespace);
            logger.trace ("===================================================================");
        }

        remoteServices.setServiceRouteTable (table);
        remoteServices.setBroadcast (broadcaster);
        if (rsd != null)
            remoteServices.setAutowireSupported (rsd.autowireRemoteServices ());

        ServiceRouteFrame srf = buildSRF (context, rsd);
        srf.udpPort = senderPort;

        // 初始化 IServiceResolver
        resolver.setLocalNamespace (namespace);
        resolver.setContext (context);

        // 启动 CAI 的 JMX 服务
        invocation = new CrossApplicationInvocation ();
        invocation.setResolver (resolver);

        // 启动路由广播服务
        routeBroadcaster = RemoteServiceFactory.getRouteBroadcaster (resolver, srf.udpPort);
        routeBroadcaster.setLocalNamespace (namespace);
        routeBroadcaster.bind (routeBroadcastPort);

        // 启动心跳管理程序
        heartbeat.setResolver (resolver);
        heartbeat.setBroadcaster (routeBroadcaster);
        heartbeat.setSendInterval (sendInterval);
        heartbeat.setScanInterval (scanInterval);
        heartbeat.setTimeout (timeout);
        heartbeat.setTable (table);
        heartbeat.start ();

        if (!srf.items.isEmpty ()) {
            table.add (srf);
        }
        receiver = new SRFBroadcastReceiver (remoteServices);
        receiver.setBroadcaster (broadcaster);
        if (rsd != null) {
            receiver.setBuildLocalStub (rsd.buildLocalStub ());
            if (rsd.ignores ().length > 0) {
                if (logger.isTraceEnabled ()) {
                    logger.trace ("ignored namespace: {}", Arrays.toString (rsd.ignores ()));
                }
                receiver.addIgnoredSRF (rsd.ignores ());
            }
            if (rsd.requires ().length > 0) {
                if (logger.isTraceEnabled ()) {
                    logger.trace ("required namespace: {}", Arrays.toString (rsd.requires ()));
                }
                receiver.addRequiredSRF (rsd.requires ());
            }
        }
        broadcaster.register (Const.Category.CAT_SRF_EVENT, receiver);
        table.setBroadcaster (broadcaster);
        executor.execute (broadcaster);

        if (!srf.items.isEmpty ()) {
            // 启动路由远程服务提供程序
/*
            provider = new SimpleProvider ();
            provider.setServiceResolver (resolver);
*/
            provider = RemoteServiceFactory.getServiceProvider (resolver);
            provider.bind (providerPort);
            // 在局域网内广播本地 srf
            routeBroadcaster.update (srf);
        } else if (logger.isTraceEnabled ()) {
            logger.trace ("there's no local service frame, do nothing");
        }
    }

    private void patchContext (SimpleObjectContext context, ServiceRouteFrame srf) throws SRFException {
        for (String beanName : context.getAllBeanNames ()) {
            Object o = context.getBean (beanName);
            Class<?> type = o.getClass ();
            Misc.buildLocalSRF (type, srf, o, beanName);
        }
    }

    private ServiceRouteFrame buildSRF (SimpleObjectContext context, ARemoteServiceDiscoverer rsd) throws Exception {
        ServiceRouteFrame srf = new ServiceRouteFrame (namespace, "127.0.0.1", providerPort);
        // 扫描已配置到托管容器内的实例
        patchContext (context, srf);
        // 如果 ARemoteServiceDiscoverer 也配置了扫描包名
        if (rsd != null && rsd.scanBasePackages ().length > 0) {
            RemoteServiceScanner scanner = new RemoteServiceScanner ();
            scanner.scan (rsd.scanBasePackages ());

            Set<Class<?>> types = scanner.getRemoteServiceClasses ();
            if (!types.isEmpty ()) {
                for (Class<?> type : types) {
                    Object bean = context.getBean (type);
                    if (bean == null) { // 只有未配置到托管容器内的才在这里重新配置
                        ARemoteService rs = type.getAnnotation (ARemoteService.class);
                        String endpoint = rs.endpoint ();
                        if (StringUtil.isEmpty (endpoint)) {
                            endpoint = Misc.getSimpleName (type);
                        }

                        Object o = type.newInstance ();
                        context.register (endpoint, o);                 // 先配置到托管容器内
                        Misc.buildLocalSRF (type, srf, o, endpoint);
                    }
                }
            }
        }

        return srf;
    }
}
